## Чтение сообщений с ручной фиксацией смещения

```1c
Процедура Слушать() Экспорт

	Инстанс = "Integration";
	РазрешеноСлушать = Константы.Слушать.Получить();

	Попытка
		Компонента = Новый(СтрШаблон("AddIn.%1.simpleKafka1C", Инстанс));   
	Исключение
		Подключено = ПодключитьВнешнююКомпоненту("/home/shmell/Source/Simple-Kafka_Adapter/build/libSimpleKafka1C.so", Инстанс, ТипВнешнейКомпоненты.Native, ТипПодключенияВнешнейКомпоненты.Изолированно);

		Если Подключено Тогда
			Компонента = Новый(СтрШаблон("AddIn.%1.simpleKafka1C", Инстанс));  	
		КонецЕсли;

	КонецПопытки;

	Если Компонента = Неопределено Тогда
		Возврат;	
	КонецЕсли;

	// установка параметров
	// для ручного управления смещением:
	// 	- enable.auto.commit = false, enable.auto.offset.store = false 
	Компонента.УстановитьПараметр("group.id", "testCompUbuntu2");   
	Компонента.УстановитьПараметр("enable.auto.commit", "false");     
	Компонента.УстановитьПараметр("enable.auto.offset.store", "false"); 
	Компонента.УстановитьПараметр("enable.partition.eof", "false");

	Компонента.КаталогЛогов = "/home/shmell/tmp/";

	Брокер = "192.168.100.101:9092";
	Топики = "testTopic";

	Результат = Компонента.ИнициализироватьКонсьюмера(Брокер, Топики); 

	// установка таймаута для ожидания сообщений - 5 сек. 
	Компонента.УстановитьТаймаутОжидания(5000);	

	Если Не Результат Тогда  

		ТекстОшибки = СтрШаблон("Не удалось инициализировать консьюмера для топиков: %1", Топики);  
		ЗаписьЖурналаРегистрации("Интеграция Kafka. Consumer", УровеньЖурналаРегистрации.Ошибка,,, ТекстОшибки);
		Возврат;

	КонецЕсли;                 

	Пока РазрешеноСлушать Цикл

		РазрешеноСлушать = Константы.Слушать.Получить();

		Если Не РазрешеноСлушать Тогда
			Прервать;	
		КонецЕсли;			      

		Попытка
			Сообщение = Компонента.Слушать(); 
		Исключение                	
			Компонента = Неопределено;                                                         
			ВызватьИсключение ОписаниеОшибки();
		КонецПопытки;	

		Если НЕ ЗначениеЗаполнено(Сообщение) Тогда
			Продолжить;
		КонецЕсли;

		ОбъектЧтение = Новый ЧтениеJSON;
		ОбъектЧтение.УстановитьСтроку(Сообщение);
		СтруктураJSON = ПрочитатьJSON(ОбъектЧтение);
		ОбъектЧтение.Закрыть();       

		ВременнаяМетка = Число(СтруктураJSON.timestamp);	
		Смещение = Число(СтруктураJSON.offset);
        Партиция = Число(СтруктураJSON.partition);

		Запись = РегистрыСведений.ДанныеКафки.СоздатьМенеджерЗаписи();
		Запись.ВременнаяМетка = ВременнаяМетка;
		Запись.Партиция = Партиция;
		Запись.Смещение = Смещение;
		Запись.Сообщение = СтруктураJSON.message;  
		Запись.ИсходноеСообщение = Сообщение; 
		Запись.Записать();                

		// вручную фиксируем смещения
		// !!! второй параметр - смещение, должно быть на 1 больше текушего прочитанного смещения
		Компонента.ЗафиксироватьСмещение(СтруктураJSON.topic, Смещение + 1, Партиция);

	КонецЦикла;  

	Компонента.ОстановитьКонсьюмера();	
	Компонента = Неопределено;	

КонецПроцедуры
```